package jk.fox.registry.discovery;

import com.alibaba.fastjson.JSON;
import jk.fox.config.registry.zookeeper.ZookeeperConfiguration;
import jk.fox.engine.RpcNodeDefinition;
import jk.fox.registry.zookeeper.ZookeeperClient;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.log4j.Logger;

import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author cuichao
 * @Description: ZookeeperDiscovery
 * @Date: create in 2020/4/17 23:48
 */
public class ZookeeperDiscovery implements NodeDiscovery<DiscoveryEvent> {

    private static Logger logger = Logger.getLogger(ZookeeperDiscovery.class);

    private ConcurrentHashMap<String, DiscoveryListener> observers = new ConcurrentHashMap<>(32);
    private ZookeeperClient zookeeper;
    private ZookeeperConfiguration configuration;

    public ZookeeperDiscovery(ZookeeperConfiguration config,ZookeeperClient zookeeper){
        this.configuration = config;
        this.zookeeper = zookeeper;
        logger.debug("ZookeeperDiscovery has instance");
    }


    @Override
    public void subscribe(DiscoveryListener listener) {
        observers.put(listener.getId(),listener);
    }

    @Override
    public DiscoveryInfo getAndSubscribe(DiscoveryListener listener) {
        observers.put(listener.getId(),listener);
        try {
            List<String> addresses = getNodes(zookeeper.getServiceNodePath(listener.getId()));
            RpcNodeDefinition definition = getDefinition(zookeeper.getServiceConfigPath(listener.getId()));
            return new DiscoveryInfo(addresses,definition);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    @Override
    public void publish(DiscoveryEvent event) {
        DiscoveryListener listener = observers.get(event.getDomain());
        if(listener != null){
            listener.listen(event);
        }
    }

    @Override
    public void discovery() {
        for (String key : observers.keySet()) {
            String path = zookeeper.getServiceNodePath(key);
            try {
                discovery(key,path);
            }catch (Exception  e){
                e.printStackTrace();
            }
        }
    }

    private RpcNodeDefinition getDefinition(String path) throws Exception {
        byte[] bytes = zookeeper.get().getData().forPath(path);
        return  JSON.parseObject(bytes, RpcNodeDefinition.class);
    }

    private List<String> getNodes(String path) throws Exception {
        List<String> children = zookeeper.get().getChildren().forPath(path);
        return children;
    }


    private void discovery(String key,String path) throws Exception {
        //监听节点
        PathChildrenCache cache = new PathChildrenCache(zookeeper.get(), path, true);
        //添加监听
        cache.getListenable().addListener((framework, event) -> {
            DiscoveryEvent discoveryEvent = null;
            switch (event.getType()) {
                case CHILD_ADDED:
                    discoveryEvent = new DiscoveryEvent(key,event.getData().getPath(),new String(event.getData().getData()),DiscoveryEvent.DiscoveryEventType.ADD_NODE);
                    break;
                case CHILD_UPDATED:
                    discoveryEvent = new DiscoveryEvent(key,event.getData().getPath(),new String(event.getData().getData()),DiscoveryEvent.DiscoveryEventType.UPDATE_NODE);
                    break;
                case CHILD_REMOVED:
                    discoveryEvent = new DiscoveryEvent(key,event.getData().getPath(),new String(event.getData().getData()),DiscoveryEvent.DiscoveryEventType.REMOVE_NODE);
                    break;
                default:
                    break;
            }
            if(discoveryEvent != null ){
                logger.debug("the server node has changed. the discoveryEvent is "+discoveryEvent.getType() +" the path is " +discoveryEvent.getData() + " for domain "+discoveryEvent.getDomain());
                publish(discoveryEvent);
            }
        });
        cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
    }


   /* private String buildPath(String key){
        return  ZookeeperClient.RPC_ROOT_PATH + "/" + key + ZookeeperClient.RPC_NODE_LIST_PATH;
    }
*/

}
